package fun.ticsmyc.rpc.common.serializer.impl;

import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;
import fun.ticsmyc.rpc.common.entity.RpcRequest;
import fun.ticsmyc.rpc.common.entity.RpcResponse;
import fun.ticsmyc.rpc.common.serializer.SerializerCode;
import fun.ticsmyc.rpc.common.exception.SerializeException;
import fun.ticsmyc.rpc.common.serializer.Serializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;

/**
 * Kryo 可以把对象信息直接写到序列化数据里，反序列化的时候可以精确地找到原始类信息，不会出错
 * 写readxxx方法时，无需传入Class或Type类信息。
 *
 * kryo提供两种读写方式。记录类型信息的writeClassAndObject/readClassAndObject方法，
 * 以及传统的writeObject/readObject方法。
 *
 * kryo的对象本身不是线程安全的，所以可以使用ThreadLocal或者kryo提供的pool
 *
 * 【kryo虽然效率高，但是只能在java中使用】
 * @author Ticsmyc
 * @date 2020-10-26 17:36
 */
public class KryoSerializer implements Serializer {

    public final byte ID = SerializerCode.KRYO.getCode();

    private static final Logger logger = LoggerFactory.getLogger(KryoSerializer.class);

    private final ThreadLocal<Kryo> kryoThreadLocal = new ThreadLocal<Kryo>(){
        @Override
        protected Kryo initialValue() {
            Kryo kryo = new Kryo();
            kryo.register(RpcResponse.class);
            kryo.register(RpcRequest.class);

            // 支持对象循环引用（否则会栈溢出），会导致性能些许下降 T_T
            kryo.setReferences(true);//默认值就是 true，添加此行的目的是为了提醒维护者，不要改变这个配置
            // 关闭序列化注册，会导致性能些许下降，但在分布式环境中，注册类生成ID不一致会导致错误
            kryo.setRegistrationRequired(false);
            // 设置类加载器为线程上下文类加载器（如果Processor来源于容器，必须使用容器的类加载器，否则妥妥的CNF）
            kryo.setClassLoader(Thread.currentThread().getContextClassLoader());
            return kryo;
        }
    };

    public KryoSerializer(){}

    /**
     * Kryo线程不安全， 不能做成单例。
     * 这个方法是为了和其他序列化器统一逻辑
     * @return
     */
    public static KryoSerializer getInstance(){
        return new KryoSerializer();
    }

    @Override
    public byte[] serialize(Object obj) {
        Output output=null;
        try{
            output = new Output(new ByteArrayOutputStream());
            Kryo kryo = kryoThreadLocal.get();
            kryo.writeObject(output,obj);
            return output.toBytes();
        } catch (Exception e) {
            logger.error("序列化时发生错误:",e);
            throw new SerializeException("序列化时有错误发生");
        }finally{
            if(output!= null){
                try{
                    output.close();
                }catch(Exception e){
                    e.printStackTrace();
                }
            }
        }
    }

    @Override
    public Object deserialize(byte[] bytes, Class<?> clazz) {
        Input input = null;
        try{
            input = new Input(new ByteArrayInputStream(bytes));
            Kryo kryo = kryoThreadLocal.get();
            return kryo.readObject(input, clazz);
        }catch(Exception e){
            logger.error("反序列化时发生错误:",e);
            throw new SerializeException("反序列化时有错误发生");
        }finally{
            if(input != null){
                try{
                    input.close();
                }catch(Exception e){
                    e.printStackTrace();
                }
            }
        }
    }


    @Override
    public byte getId() {
        return ID;
    }
}
